package cn.yanceysong.cdc;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class MysqlCdcDemoMain {
    private static final Logger log = LoggerFactory.getLogger(MysqlCdcDemoMain.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置 Redis 连接
        FlinkJedisPoolConfig jedisPoolConfig = new FlinkJedisPoolConfig.Builder()
                .setHost("127.0.0.1")
                .setPort(6379)
                .setDatabase(0)
                .setPassword("123456")
                .build();

        // 创建 Redis Sink
        RedisSink<String> redisSink = new RedisSink<>(jedisPoolConfig, new RedisKVMapper());

        Properties properties = new Properties();
        properties.put("useUnicode", "true");
        properties.put("characterEncoding", "utf8");
        properties.put("useSSL", "false");

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("127.0.0.1")
                .port(3306)
                .databaseList("flink_test") // 设置捕获的数据库， 如果需要同步整个数据库，请将 tableList 设置为 ".*".
                .tableList("flink_test.users") // 设置捕获的表
                .username("root")
                .password("123456")
                .serverTimeZone("Asia/Shanghai")  // 设置 MySQL 服务器的时区
                .jdbcProperties(properties)
                .startupOptions(StartupOptions.earliest())
                .deserializer(new JsonDebeziumDeserializationSchema()) // 将 SourceRecord 转换为 JSON 字符串
                .build();


        // 设置 3s 的 checkpoint 间隔
        env.enableCheckpointing(3000);
        SingleOutputStreamOperator<String> outSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
                // 设置 source 节点的并行度为 4
                .setParallelism(1);
        outSource.addSink(redisSink);
        env.execute("Print MySQL Snapshot + Binlog");
    }
}